package com.lost.octopus.es.processor.mq.consumer.dto;

import com.lost.octopus.es.processor.entity.db.EsDataCaptureRuleEntity;
import com.lost.octopus.es.processor.scheduler.EsSyncIndexOperator;
import com.sunsharing.share.boot.framework.AppUtils;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;

import java.util.LinkedHashSet;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 处理数据捕获的对象
 * @author zhangbh
 * @date 2025/4/23 10:58
 */
@Log4j2
@Getter
public class SinkDeal {

    private final String id;

    /**
     * 数据库
     */
    private final String acceptSchema;

    /**
     * 表名
     */
    private final String acceptTable;

    /**
     * 查询sql
     */
    private final String sql;
    /**
     * 数据处理操作者
     */
    @Setter
    private EsSyncIndexOperator<?, ?, ?, ?> opt;

    /**
     * in参数，注册的时候解析sql，分析出的参数名
     */
    @Setter(AccessLevel.PRIVATE)
    private String inParamName;

    /**
     * 子规则，通过父规则查询出来的id，再进行in查询
     */
    private final Set<SinkDeal> children = new LinkedHashSet<>();

    // ==================================== 打印日志/初始化使用的 ====================================

    /**
     * es主键类型
     */
    @Setter
    private Class<?> idType;
    // ==================================== 打印日志/初始化使用的 ====================================

    public SinkDeal(EsDataCaptureRuleEntity entity, boolean hasSql) {
        String optBeanName = entity.getEsOperatorBeanName();
        // optBeanName首字母小写
        optBeanName = StringUtils.uncapitalize(optBeanName);
        String sql = entity.getQuerySql();
        String acceptTable = entity.getAcceptTable();
        String desc = entity.getRuleId();
        if (hasSql && StringUtils.isBlank(sql)) {
            throw new IllegalArgumentException("注册的变更捕获没有sql！！！ desc=" + desc);
        }
        this.id = entity.getRuleId();
        Object bean;
        try {
            bean = AppUtils.getBean(optBeanName);
        } catch (Exception e) {
            throw new IllegalArgumentException("注册的变更捕获配置获取不到es同步类 ！！！ desc=" + desc);
        }
        boolean isEsBean = bean instanceof EsSyncIndexOperator;
        if (!isEsBean) {
            throw new IllegalArgumentException(String.format("注册的es同步类%s非EsSyncIndexOperator ！！！ desc=%s", optBeanName, desc));
        }
        this.opt = (EsSyncIndexOperator<?, ?, ?, ?>) bean;
        this.idType = this.opt.idType();
        this.acceptSchema = entity.getAcceptSchema();
        this.acceptTable = acceptTable;
        this.sql = sql;
    }

    /**
     * 加入子规则，优点，基于parent的id进行数据二次处理，对数据库和sql复杂度友好
     * 例如：数据dz变更，添加 ld sink，这时候可以 ld sink增加一个fw的children
     * @param hasSql 是否需要sql，若有，则默认按parentId进行 sql in查询，否则直接使用parentId刷新es索引
     * @return childrenSink
     */
    public SinkDeal childSink(EsDataCaptureRuleEntity entity, boolean hasSql) {
        SinkDeal info = new SinkDeal(entity, hasSql);
        if (hasSql) {
            String searchEsIdSql = info.getSql();
            // 通过正则  \(:(.*)\) 获取配置名
            Matcher matcher = Pattern.compile("\\(:(.*)\\)").matcher(searchEsIdSql);
            if (!matcher.find()) {
                throw new IllegalArgumentException("请确认关联sql注册正确，描述：" + id + "  sql中需要附带in参数占位符！！！！  如:dzbh");
            }
            info.setInParamName(matcher.group(1));
            log.info("{} 初始化子配置 参数 {}", id, info.getInParamName());
        }
        children.add(info);
        return info;
    }

    public String getDesc() {
        return id;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        SinkDeal info = (SinkDeal) o;
        return Objects.equals(sql, info.sql) && Objects.equals(idType, info.idType) && Objects.equals(id, info.id);
    }

    @Override
    public int hashCode() {
        return Objects.hash(sql, idType, id);
    }
}
